package com.itheima.wordcount;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class WordCountMain extends Configured implements Tool {
    @Override
    public int run(String[] args) throws Exception {
        //天龙八部
        //1、获得job任务对象
        Job job = Job.getInstance(super.getConf());

        //在yarn平台运行的必备参数
        job.setJarByClass(WordCountMain.class);
        //2、封装天龙八部
        //2.1
        job.setInputFormatClass(TextInputFormat.class);
        FileInputFormat.addInputPath(job,new Path(args[0]));
        //2.2设置mapTask组件
        job.setMapperClass(WordCountMapperTask.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        //2.3设置shuffle的操作：分区，排序 规约 分组  默认即可

        //2.7设置reduceTask的组件，reduce类和reduce的输出的k3和v3的类型
        job.setReducerClass(WordCountReduceTask.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        //2.8输出组件
        job.setOutputFormatClass(TextOutputFormat.class);
        FileOutputFormat.setOutputPath(job,new Path(args[1]));
        //3.提交任务
        boolean flag = job.waitForCompletion(true);//如果任务成功，返回true

        return flag ? 0:1;
    }

    public static void main(String[] args) throws Exception {
        //1、基于tool调用run方法
        Configuration conf = new Configuration();
        int i = ToolRunner.run(conf, new WordCountMain(), args); //0正常  1异常退出

        //退出程序
        System.exit(i);
    }
}
